package com.shf.dynamodb;

import java.io.File;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
import com.amazonaws.services.dynamodbv2.document.DynamoDB;
import com.amazonaws.services.dynamodbv2.document.Item;
import com.amazonaws.services.dynamodbv2.document.ItemCollection;
import com.amazonaws.services.dynamodbv2.document.PrimaryKey;
import com.amazonaws.services.dynamodbv2.document.PutItemOutcome;
import com.amazonaws.services.dynamodbv2.document.QueryOutcome;
import com.amazonaws.services.dynamodbv2.document.ScanOutcome;
import com.amazonaws.services.dynamodbv2.document.Table;
import com.amazonaws.services.dynamodbv2.document.UpdateItemOutcome;
import com.amazonaws.services.dynamodbv2.document.spec.DeleteItemSpec;
import com.amazonaws.services.dynamodbv2.document.spec.GetItemSpec;
import com.amazonaws.services.dynamodbv2.document.spec.QuerySpec;
import com.amazonaws.services.dynamodbv2.document.spec.ScanSpec;
import com.amazonaws.services.dynamodbv2.document.spec.UpdateItemSpec;
import com.amazonaws.services.dynamodbv2.document.utils.NameMap;
import com.amazonaws.services.dynamodbv2.document.utils.ValueMap;
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.KeyType;
import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
import com.amazonaws.services.dynamodbv2.model.ReturnValue;
import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

/**
 * 基础命令测试
 * @author songhaifeng
 *
 */
public class BaseCommandDemo {

    public static void main(String[] args) throws Exception {
//    	//创建表
//    	createTable();
    	//插入数据
    	insertData();
//    	//新增一个项目
//    	addOneProject();
//    	//读取上一步新增项目
//    	readOneProject();
//    	//更新新增项目
//    	updateOneProject();
//    	//递增一个原子计数器
//    	addCount();
//    	//附有条件的情况下更新数据
//    	updateByCondition();
//    	//删除一个项目
//    	deleteOneProject();
//    	//检索所有在 1985 year 发行的电影。
//        //检索所有在 1992 year 发行并且 title 以字母“A”到字母“L”开头的电影。
//    	query();
//    	//扫描整个 Movies 表，表中包含约 5000 个项目。扫描时可指定可选筛选条件，以便仅检索 20 世纪 50 年代以来的电影（约 100 个项目），同时舍弃所有其他项目
//    	scan();
//    	//删除表
//    	dropTable();
    }
    
    /**
     * 创建表
     */
    public static void createTable(){
    	DynamoDB dynamoDB=getDynamodb();

         String tableName = "Movies";

         try {
             System.out.println("Attempting to create table; please wait...");
             Table table = dynamoDB.createTable(tableName,
                 Arrays.asList(
                     new KeySchemaElement("year", KeyType.HASH),  //Partition key
                     new KeySchemaElement("title", KeyType.RANGE)), //Sort key
                     Arrays.asList(
                         new AttributeDefinition("year", ScalarAttributeType.N),
                         new AttributeDefinition("title", ScalarAttributeType.S)), 
                     new ProvisionedThroughput(10L, 10L));
             table.waitForActive();
             System.out.println("Success.  Table status: " + table.getDescription().getTableStatus());

         } catch (Exception e) {
             System.err.println("Unable to create table: ");
             System.err.println(e.getMessage());
         }finally{
        	 	dynamoDB.shutdown();
         }
    }
    
    /**
     * 插入数据
     * @throws Exception
     */
    public static void insertData() throws Exception {

    	DynamoDB dynamoDB=getDynamodb();

        Table table = dynamoDB.getTable("Movies");

        JsonParser parser = new JsonFactory()
            .createParser(new File("D:\\learnworkspace\\dynamodbdemo\\src\\main\\java\\moviedata.json"));

        JsonNode rootNode = new ObjectMapper().readTree(parser);
        Iterator<JsonNode> iter = rootNode.iterator();

        ObjectNode currentNode;

        while (iter.hasNext()) {
            currentNode = (ObjectNode) iter.next();

            int year = currentNode.path("year").asInt();
            String title = currentNode.path("title").asText();

            try {
                table.putItem(new Item()
                .withPrimaryKey("year", year, "title", title)
                .withJSON("info", currentNode.path("info").toString()));
                System.out.println("PutItem succeeded: " + year + " " + title);

            } catch (Exception e) {
                System.err.println("Unable to add movie: " + year + " " + title);
                System.err.println(e.getMessage());
                break;
            }
        }
        parser.close();
   	 	dynamoDB.shutdown();
    }
    
    public static void addOneProject() throws Exception {
    	DynamoDB dynamoDB=getDynamodb();
        
        Table table = dynamoDB.getTable("Movies");

        int year = 2015;
        String title = "The Big New Movie";
        
        final Map<String, Object> infoMap = new HashMap<String, Object>();
        infoMap.put("plot",  "Nothing happens at all.");
        infoMap.put("rating",  0);

        try {
            System.out.println("Adding a new item...");
            PutItemOutcome outcome = table.putItem(new Item()
                .withPrimaryKey("year", year, "title", title)
                .withMap("info", infoMap));

            System.out.println("PutItem succeeded:\n" + outcome.getPutItemResult());

        } catch (Exception e) {
            System.err.println("Unable to add item: " + year + " " + title);
            System.err.println(e.getMessage());
        }finally{
       	 	dynamoDB.shutdown();
        }

    }
    
    /**
     * 读取上一步新增项目
     * @throws Exception
     */
    public static void readOneProject() throws Exception {
    	DynamoDB dynamoDB=getDynamodb();

        Table table = dynamoDB.getTable("Movies");

        int year = 2015;
        String title = "The Big New Movie";
        
        GetItemSpec spec = new GetItemSpec()
            .withPrimaryKey("year", year, "title", title);

        try {
            System.out.println("Attempting to read the item...");
            Item outcome = table.getItem(spec);
            System.out.println("GetItem succeeded: " + outcome);

        } catch (Exception e) {
            System.err.println("Unable to read item: " + year + " " + title);
            System.err.println(e.getMessage());
        }finally{
       	 	dynamoDB.shutdown();
        }
    }
    
    /**
     * 更新之前新增的项目
     * @param args
     * @throws Exception
     */
    public static void updateOneProject() throws Exception {
    	DynamoDB dynamoDB=getDynamodb();

        Table table = dynamoDB.getTable("Movies");

        int year = 2015;
        String title = "The Big New Movie";

        UpdateItemSpec updateItemSpec = new UpdateItemSpec()
            .withPrimaryKey("year", year, "title", title)
            .withUpdateExpression("set info.rating = :r, info.plot=:p, info.actors=:a")
            .withValueMap(new ValueMap()
                .withNumber(":r", 5.5)
                .withString(":p", "Everything happens all at once.")
                .withList(":a", Arrays.asList("Larry","Moe","Curly")))
            .withReturnValues(ReturnValue.UPDATED_NEW);

        try {
            System.out.println("Updating the item...");
            UpdateItemOutcome outcome = table.updateItem(updateItemSpec);
            System.out.println("UpdateItem succeeded:\n" + outcome.getItem().toJSONPretty());
            
        } catch (Exception e) {
            System.err.println("Unable to update item: " + year + " " + title);
            System.err.println(e.getMessage());
        }finally{
       	 	dynamoDB.shutdown();
        }
    }
    
    /**
     * 递增一个原子计数器
     * @param args
     * @throws Exception
     */
    public static void addCount() throws Exception {

    	DynamoDB dynamoDB=getDynamodb();

        Table table = dynamoDB.getTable("Movies");

        int year = 2015;
        String title = "The Big New Movie";

        UpdateItemSpec updateItemSpec = new UpdateItemSpec()
            .withPrimaryKey("year", year, "title", title)
            .withUpdateExpression("set info.rating = info.rating + :val")
            .withValueMap(new ValueMap()
            .withNumber(":val", 1))
            .withReturnValues(ReturnValue.UPDATED_NEW);

        try {
            System.out.println("Incrementing an atomic counter...");
            UpdateItemOutcome outcome = table.updateItem(updateItemSpec);
            System.out.println("UpdateItem succeeded:\n" + outcome.getItem().toJSONPretty());
            
        } catch (Exception e) {
            System.err.println("Unable to update item: " + year + " " + title);
            System.err.println(e.getMessage());
        }finally{
       	 	dynamoDB.shutdown();
        }
    }
    
    /**
     * 附有条件更新数据
     * @param args
     * @throws Exception
     */
    public static void updateByCondition() throws Exception {

    	DynamoDB dynamoDB=getDynamodb();

        Table table = dynamoDB.getTable("Movies");

        int year = 2015;
        String title = "The Big New Movie";

        UpdateItemSpec updateItemSpec = new UpdateItemSpec()
            .withPrimaryKey(new PrimaryKey("year", year, "title", title))
            .withUpdateExpression("remove info.actors[0]")
//          //需要大于3名演员方可以更新--无法更新不满足条件
//            .withConditionExpression("size(info.actors) > :num")
            //满足大于等于即可更新--更新成功
            .withConditionExpression("size(info.actors) >= :num")
            .withValueMap(new ValueMap().withNumber(":num", 3))
            .withReturnValues(ReturnValue.UPDATED_NEW);

        try {
            System.out.println("Attempting a conditional update...");
            UpdateItemOutcome outcome = table.updateItem(updateItemSpec);
            System.out.println("UpdateItem succeeded:\n" + outcome.getItem().toJSONPretty());

        } catch (Exception e) {
            System.err.println("Unable to update item: " + year + " " + title);
            System.err.println(e.getMessage());
        }finally{
       	 	dynamoDB.shutdown();
        }
    }
    
    /**
     * 删除项目
     * @param args
     * @throws Exception
     */
    public static void deleteOneProject() throws Exception {

    	DynamoDB dynamoDB=getDynamodb();

        Table table = dynamoDB.getTable("Movies");

        int year = 2015;
        String title = "The Big New Movie";

        DeleteItemSpec deleteItemSpec = new DeleteItemSpec()
            .withPrimaryKey(new PrimaryKey("year", year, "title", title))
//            //不满足条件，更新失败，当前值大于5.0
//            .withConditionExpression("info.rating <= :val")
            //满足条件，更新成功，当前值大于5.0
            .withConditionExpression("info.rating >= :val")
            .withValueMap(new ValueMap()
            .withNumber(":val", 5.0));
        
        try {
            System.out.println("Attempting a conditional delete...");
            table.deleteItem(deleteItemSpec);
            System.out.println("DeleteItem succeeded");
        } catch (Exception e) {
            System.err.println("Unable to delete item: " + year + " " + title);
            System.err.println(e.getMessage());
        }finally{
        	 dynamoDB.shutdown();
        }
    }
    
    /**
     * 检索所有在 1985 year 发行的电影。
     * 检索所有在 1992 year 发行并且 title 以字母“A”到字母“L”开头的电影。
     * @param args
     * @throws Exception
     */
    public static void query() throws Exception {
    	
    	DynamoDB dynamoDB=getDynamodb();

        Table table = dynamoDB.getTable("Movies");

        //nameMap 提供名称替换功能。我们使用此参数是因为 year 是 DynamoDB 中的保留字，您不能直接在任何表达式中使用它，包括 KeyConditionExpression。我们使用表达式属性名称 #yr 来解决此问题。
        HashMap<String, String> nameMap = new HashMap<String, String>();
        nameMap.put("#yr", "year");

        //valueMap 提供值替换功能。我们使用此参数是因为您不能在任何表达式中使用文字，包括 KeyConditionExpression。我们使用表达式属性值 :yyyy 来解决此问题。
        HashMap<String, Object> valueMap = new HashMap<String, Object>();
        valueMap.put(":yyyy", 1985);
        
        //检索所有在 1985 year 发行的电影。
        QuerySpec querySpec = new QuerySpec()
            .withKeyConditionExpression("#yr = :yyyy")
            .withNameMap(nameMap)
            .withValueMap(valueMap);

        ItemCollection<QueryOutcome> items = null;
        Iterator<Item> iterator = null;
        Item item = null;

        try {
            System.out.println("Movies from 1985");
            items = table.query(querySpec);

            iterator = items.iterator();
            while (iterator.hasNext()) {
                item = iterator.next();
                System.out.println(item.getNumber("year") + ": "
                        + item.getString("title"));
            }

        } catch (Exception e) {
            System.err.println("Unable to query movies from 1985");
            System.err.println(e.getMessage());
        }

        valueMap.put(":yyyy", 1992);
        valueMap.put(":letter1", "A");
        valueMap.put(":letter2", "L");

        //检索所有在 1992 year 发行并且 title 以字母“A”到字母“L”开头的电影。
        querySpec
                .withProjectionExpression(
                        "#yr, title, info.genres, info.actors[0]")
                .withKeyConditionExpression(
                        "#yr = :yyyy and title between :letter1 and :letter2")
                .withNameMap(nameMap).withValueMap(valueMap);

        try {
            System.out
                    .println("Movies from 1992 - titles A-L, with genres and lead actor");
            items = table.query(querySpec);

            iterator = items.iterator();
            while (iterator.hasNext()) {
                item = iterator.next();
                System.out.println(item.getNumber("year") + ": "
                        + item.getString("title") + " " + item.getMap("info"));
            }

        } catch (Exception e) {
            System.err.println("Unable to query movies from 1992:");
            System.err.println(e.getMessage());
        }finally{
       	 	dynamoDB.shutdown();
    	}
    }
    
    /**
     * scan扫描表:扫描整个 Movies 表，表中包含约 5000 个项目。扫描时可指定可选筛选条件，以便仅检索 20 世纪 50 年代以来的电影（约 100 个项目），同时舍弃所有其他项目
     * @param args
     * @throws Exception
     */
    public static void scan() throws Exception {

    	DynamoDB dynamoDB=getDynamodb();
    	
        Table table = dynamoDB.getTable("Movies");

        ScanSpec scanSpec = new ScanSpec()
        	//ProjectionExpression 用于指定要在扫描结果中包含的属性。
            .withProjectionExpression("#yr, title, info.rating")
            //FilterExpression 用于指定一个条件，以便仅返回符合条件的项目。所有其他项目都将被舍弃
            .withFilterExpression("#yr between :start_yr and :end_yr")
            .withNameMap(new NameMap().with("#yr",  "year"))
            .withValueMap(new ValueMap().withNumber(":start_yr", 1950).withNumber(":end_yr", 1959));

        try {
            ItemCollection<ScanOutcome> items = table.scan(scanSpec);

            Iterator<Item> iter = items.iterator();
            while (iter.hasNext()) {
                Item item = iter.next();
                System.out.println(item.toString());
            }
            
        } catch (Exception e) {
            System.err.println("Unable to scan the table:");
            System.err.println(e.getMessage());
        }finally {
        	dynamoDB.shutdown();
		}
    }
    
    //删除表
    public  static void dropTable(){
    	DynamoDB dynamoDB=getDynamodb();

        Table table = dynamoDB.getTable("Movies");

        try {
            System.out.println("Attempting to delete table; please wait...");
            table.delete();
            table.waitForDelete();
            System.out.print("Success.");

        } catch (Exception e) {
            System.err.println("Unable to delete table: ");
            System.err.println(e.getMessage());
        }finally {
        	dynamoDB.shutdown();
		}
    }
    
    /**
     * 建立连接
     * @return
     */
    public static DynamoDB getDynamodb(){
    	AmazonDynamoDB client = AmazonDynamoDBClientBuilder.standard().withEndpointConfiguration(
                new AwsClientBuilder.EndpointConfiguration("http://localhost:8765", "us-west-2"))
                .build();

        DynamoDB dynamoDB = new DynamoDB(client);

        return dynamoDB;
    }
    
}